package com.link.serve.kafka.producer;

import com.link.serve.constants.LinkRedisKey;
import com.link.serve.kafka.factory.DynamicKafkaProducerFactory;
import com.link.serve.kafka.proto.KafkaMessageProto;
import com.link.serve.tcp.IMServerGroup;
import com.link.serve.tcp.message.proto.PackHeader;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import java.util.*;
@Component
@Slf4j
public class KafkaMessageProducer {
    @Value("${redis.kafka.producer.key-serializer}")//配置业务指定的序列化器全限定类名
    private String redisKeySerializerClassName;
    @Value("${redis.kafka.producer.value-serializer}")
    private String redisValueSerializerClassName;

    @Value("${spring.kafka.bootstrap-servers}")//配置指定的redis集群地址
    private String bootstrapServers;
    public Boolean sendRedisMessage(PackHeader.Header header, String redisType, String messageType) throws ClassNotFoundException {
        return sendMessage(
                header.getUid(),
                redisType,
                IMServerGroup.serverId,
                String.join(":", LinkRedisKey.LINK_USER_SERVER_ID, String.valueOf(header.getUid())),
                messageType
        );
    }
    public Boolean sendMessage(Long userId, String operationType, Long serverId, String key, String message) throws ClassNotFoundException {

        String kafkaTopic = "user-topic";// 设定 Topic，不同 topic 可以有不同的序列化配置

        Class<?> keySerializer = Class.forName(redisKeySerializerClassName);
        Class<?> valueSerializer = Class.forName(redisValueSerializerClassName);

        Map<String, Object> extraConfigs = new HashMap<>();// 若有额外配置，可组装到 extraConfigs 中，否则传 null

        KafkaTemplate<String, KafkaMessageProto.KafkaMessage> kafkaTemplate =// 根据构造好的 ProducerFactory 创建 KafkaTemplate
                new KafkaTemplate<>(
                    DynamicKafkaProducerFactory
                    .createProducerFactory(
                    bootstrapServers, keySerializer, valueSerializer, extraConfigs));

        KafkaMessageProto.KafkaMessage kafkaMessage = KafkaMessageProto.KafkaMessage.newBuilder()// 构造 KafkaMessageProto 消息
                .setOperationType(operationType)
                .setUserId(userId)
                .setServerId(serverId)
                .setKey(key)
                .setMessage(message)
                .build();

        ProducerRecord<String, KafkaMessageProto.KafkaMessage> record =// 注意：取消传入 partition，由 Kafka 默认根据 key 的 hash 分区
                new ProducerRecord<>(kafkaTopic, key, kafkaMessage);

        ListenableFuture<SendResult<String, KafkaMessageProto.KafkaMessage>> future = kafkaTemplate.send(record);// 发送消息

        future.addCallback(new ListenableFutureCallback<SendResult<String, KafkaMessageProto.KafkaMessage>>() {
            @Override
            public void onSuccess(SendResult<String, KafkaMessageProto.KafkaMessage> result) {
                log.info("msg_send_✅: type={} topic={} partition={} offset={}",
                        message, kafkaTopic, result.getRecordMetadata().partition(), result.getRecordMetadata().offset());
            }
            @Override
            public void onFailure(Throwable ex) {
                log.error("msg_send_❌: type={} topic={} userId={} error={}",
                        message, kafkaTopic, userId, ex.getMessage());
            }
        });return true;
    }
}
